/*
 * Copyright (C) 2008 Universidade Federal de Campina Grande
 *  
 * This file is part of OurGrid. 
 *
 * OurGrid is free software: you can redistribute it and/or modify it under the
 * terms of the GNU Lesser General Public License as published by the Free 
 * Software Foundation, either version 3 of the License, or (at your option) 
 * any later version. 
 * 
 * This program is distributed in the hope that it will be useful, but WITHOUT 
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or 
 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public License
 * for more details. 
 * 
 * You should have received a copy of the GNU Lesser General Public License 
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 * 
 */
package org.ourgrid.peer.controller.accounting;

import java.util.List;

import org.ourgrid.common.interfaces.PeerAccounting;
import org.ourgrid.common.interfaces.to.Accounting;
import org.ourgrid.common.interfaces.to.GridProcessAccounting;
import org.ourgrid.common.interfaces.to.RequestSpec;
import org.ourgrid.common.interfaces.to.WorkAccounting;
import org.ourgrid.common.statistics.beans.peer.Job;
import org.ourgrid.common.statistics.beans.status.ExecutionStatus;
import org.ourgrid.common.statistics.control.AccountingControl;
import org.ourgrid.common.statistics.control.JobControl;
import org.ourgrid.common.statistics.control.UserControl;
import org.ourgrid.common.statistics.control.WorkerControl;
import org.ourgrid.peer.PeerConfiguration;
import org.ourgrid.peer.controller.messages.AccountingMessages;
import org.ourgrid.peer.dao.AccountingDAO;
import org.ourgrid.peer.dao.AllocationDAO;
import org.ourgrid.peer.dao.ConsumerDAO;
import org.ourgrid.peer.dao.LocalWorkersDAO;
import org.ourgrid.peer.dao.RequestDAO;
import org.ourgrid.peer.dao.UsersDAO;
import org.ourgrid.peer.dao.statistics.PeerDAO;
import org.ourgrid.peer.to.AllocableWorker;
import org.ourgrid.peer.to.LocalWorker;
import org.ourgrid.peer.to.PeerBalance;
import org.ourgrid.peer.to.RemoteConsumer;
import org.ourgrid.reqtrace.Req;

import br.edu.ufcg.lsd.commune.api.InvokeOnDeploy;
import br.edu.ufcg.lsd.commune.container.servicemanager.ServiceManager;
import br.edu.ufcg.lsd.commune.identification.DeploymentID;
import br.edu.ufcg.lsd.commune.identification.ServiceID;
import br.edu.ufcg.lsd.commune.network.certification.CertificationUtils;

/**
 * Performs accounting actions
 */
@Req("REQ027")
public class AccountingController implements PeerAccounting{
	
	private static final long serialVersionUID = 1L;

	
	private ServiceManager serviceManager;

	@InvokeOnDeploy
	public void init(ServiceManager serviceManager) {
		this.serviceManager = serviceManager;
	}

	/**
	 * Validates if the sender can report the replica accounting and if the 
	 * accounting data is valid. Call the DAO to store the replica accounting 
	 * data.
	 * 
	 * The sender must have its public key already saved in the local users 
	 * set.
	 * 
	 * The replica accounting must contain the request specification and a
	 * valid allocable worker.
	 * 
	 * The request must be running in this peer.
	 * 
	 * @param userPublicKey The sender public key
	 * @param replicaAccounting The replica accounting data
	 * 
	 * @see UsersDAO#getUserByPublicKey(String)
	 * @see RequestDAO#isRunning(RequestSpec)
	 * @see AccountingDAO#addReplicaAccounting(Accounting)
	 */
	public void reportReplicaAccounting(GridProcessAccounting replicaAccounting) {
		String userPublicKey = serviceManager.getSenderPublicKey();
		
		UsersDAO usersDAO = serviceManager.getDAO(UsersDAO.class);
		
		if (! UserControl.getInstance().userExists(userPublicKey, usersDAO)) {
			serviceManager.getLog().warn(AccountingMessages.getUnknownUserReplicaAccountingMessage(userPublicKey));
			return;
		}
		
		if (!usersDAO.isLoggedUser(userPublicKey)) {
			serviceManager.getLog().warn(AccountingMessages.getNotLoggerUserMessage(userPublicKey));
			return;
		}
			
		DeploymentID userObjectID = serviceManager.getDAO(UsersDAO.class).getLoggedUser(userPublicKey).getDeploymentId();
		
		if (replicaAccounting == null) {
			serviceManager.getLog().warn(AccountingMessages.getNullReplicaAccountingMessage(userObjectID));
			return;
		}
	
//		if (!validateExecutionState(replicaAccounting)) {
//			return;
//		}
		
		RequestSpec requestSpec = replicaAccounting.getRequestSpec();
		if (!validateRequestSpec(userPublicKey, userObjectID, requestSpec)) {
			return;
		}
		
		if (replicaAccounting.getWorkerID() == null) {
			serviceManager.getLog().warn(AccountingMessages.getUserWithoutWorkerMessage(userObjectID));
			return;
		}

		DeploymentID workerID = new DeploymentID(replicaAccounting.getWorkerID());
		workerID.setPublicKey(replicaAccounting.getWorkerPublicKey());
		
		AllocableWorker allocableWorker = validateWorker(userPublicKey, userObjectID, workerID);
		if (allocableWorker == null) {
			return;
		}

		PeerBalance accountings = replicaAccounting.getAccountings();
		if (!validateAccountings(userObjectID, workerID, accountings)) {
			return;
		}
		
		serviceManager.getLog().debug(AccountingMessages.getReceivedReplicaAccountingMessage(
				userObjectID, requestSpec.getRequestId(), workerID,
				accountings, replicaAccounting.getState(), allocableWorker.isWorkerLocal()));
		
//		DeploymentID wpID = null;
//		Object workerProvider = allocableWorker.getWorkerProvider();
		
//		if (allocableWorker.isWorkerLocal()) {
//			wpID = serviceManager.getLocalObjectID(workerProvider);
//		} else {
//			wpID = serviceManager.getStubDeploymentID(workerProvider);
//		}
		
		AccountingControl.getInstance().addReplicaAccounting(replicaAccounting, 
				allocableWorker.getProviderCertificateDN(), serviceManager);
		
		JobControl.getInstance().addProcessAccounting(replicaAccounting, serviceManager);
	}

	private boolean validateRequestSpec(String userPublicKey, DeploymentID userObjectID, RequestSpec requestSpec) {

		if (requestSpec == null) {
			serviceManager.getLog().warn(AccountingMessages.getNullRequestMessage(userObjectID));
			return false;
		}

		Job job = JobControl.getInstance().findByRequestId(requestSpec.getRequestId(), serviceManager);
		
		if (job == null || (job.getStatus() != null && job.getStatus().equals(ExecutionStatus.FINISHED))) {
			serviceManager.getLog().warn(AccountingMessages.getInexistentRequestMessage(userObjectID, requestSpec.getRequestId()));
			return false;
		}
		
		if (!job.getLogin().getUser().getPublicKey().equals(userPublicKey)) {
			serviceManager.getLog().warn(AccountingMessages.getWrongRequestConsumerMessage(userObjectID, requestSpec.getRequestId()));
			return false;
		}
		
		return true;
	}

	private AllocableWorker validateWorker(String userPublicKey, DeploymentID userObjectID, DeploymentID workerObjectID) {
		if (workerObjectID == null) {
			serviceManager.getLog().warn(AccountingMessages.getUserWithoutWorkerMessage(userObjectID));
			return null;
		}
		
		AllocableWorker allocableWorker = 
			serviceManager.getDAO(AllocationDAO.class).getAllocableWorker(workerObjectID.getPublicKey());
		
		String publicKey = "";
		if (allocableWorker != null) {
			publicKey = allocableWorker.getConsumerPublicKey();
		}
		
		if (allocableWorker == null || !userPublicKey.equals(publicKey) || !allocableWorker.isDelivered()) {
			
			serviceManager.getLog().warn(AccountingMessages.getWorkerNotAllocatedForUserMessage(userObjectID, workerObjectID,
					allocableWorker));
			
			return null;
		}
		return allocableWorker;
	}

	private boolean validateAccountings(DeploymentID userObjectID, DeploymentID workerObjectID, PeerBalance accountings) {
		
		if (accountings.getAttribute(PeerBalance.CPU_TIME) <= 0) {
			serviceManager.getLog().warn(AccountingMessages.getNegativeCPUReplicaAccountingMessage(userObjectID, workerObjectID));
			return false;
		}
		
		if (accountings.getAttribute(PeerBalance.DATA) < 0) {
			serviceManager.getLog().warn(AccountingMessages.getNonPositiveDataReplicaAccountingMessage(userObjectID, workerObjectID));
			return false;
		}
		
		return true;
	}


	/**
	 * Validate if the sender can report the work accountings, calculate the
	 * new balances for the remote peers that consumed the worker, and call the
	 * DAO to update the remote peers balance.
	 * 
	 * The sender must have its public key already saved in the local workers
	 * set.
	 * 
	 * The AccountingEvaluator is responsible for calculating the balance from
	 * the accounting data.
	 * 
	 * @param workerPublicKey The sender public key
	 * @param consumersAccountings A list of accountings for the remote peers 
	 * 			that consumed the worker
	 * 
	 * @see LocalWorkersDAO#getRecoveredWorker(String)
	 * @see AccountingDAO#getRemotePeerBalance(String)
	 * @see AccountingDAO#setRemotePeerBalance(String, Double)
	 */
	public void reportWorkAccounting(List<WorkAccounting> consumersAccountings) {
		String workerPublicKey = serviceManager.getSenderPublicKey();
		ServiceID senderServiceID = serviceManager.getSenderServiceID();
		
		LocalWorker localWorker = WorkerControl.getInstance().getRecoveredWorker(
				senderServiceID, serviceManager.getDAO(LocalWorkersDAO.class));
				
		if (!validateWorker(workerPublicKey, localWorker)) {
			return;
		}
		
		for (WorkAccounting accounting : consumersAccountings) {

			DeploymentID wmOID = serviceManager.getStubDeploymentID(localWorker.getWorkerManagement());
			
			if (!validateAccounting(accounting, wmOID)) {
				continue;
			}
			
			serviceManager.getLog().debug(AccountingMessages.getReceivedWorkAccountingMessage(accounting, wmOID));
			
			commitReceivedFavour(accounting);
		}
	}

	private boolean validateAccounting(WorkAccounting accounting, DeploymentID workerManagOID) {
		String consumerPeerPublicKey = accounting.getConsumerPeerPublicKey();
		
		if (consumerPeerPublicKey == null) {
			serviceManager.getLog().warn(AccountingMessages.getNoConsumerWorkAccountingMessage(workerManagOID));
			return false;
		}
		
		if (accounting.getAccountings().getAttribute(PeerBalance.CPU_TIME) <= 0) {
			serviceManager.getLog().warn(AccountingMessages.getNegativeCPUWorkAccountingMessage(workerManagOID,
					consumerPeerPublicKey));
			return false;
		}
		
		if (accounting.getAccountings().getAttribute(PeerBalance.DATA) < 0) {
			serviceManager.getLog().warn(AccountingMessages.getNonPositiveDataWorkAccountingMessage(workerManagOID,
					consumerPeerPublicKey));
			return false;
		}
		
		if (serviceManager.isThisMyPublicKey(consumerPeerPublicKey)) {
			serviceManager.getLog().warn(AccountingMessages.getLocalPeerWorkAccountingMessage(workerManagOID));
			return false;
		}
		
		return true;
	}

	private boolean validateWorker(String workerPublicKey, LocalWorker localWorker) {
		if (localWorker == null) {
			serviceManager.getLog().warn(AccountingMessages.getUnknownWorkerMessage(workerPublicKey));
			return false;
		}
		
		return true;
	}

	private void commitReceivedFavour(WorkAccounting accounting) {
		String consumerPeerPublicKey = accounting.getConsumerPeerPublicKey();
		
		RemoteConsumer consumer = serviceManager.getDAO(ConsumerDAO.class).getRemoteConsumer(consumerPeerPublicKey);
		
		String remoteDNData = null;
		
		if (consumer != null) {
			remoteDNData = consumer.getConsumerDN();
		}
		
		String localDNData = CertificationUtils.getCertSubjectDN(serviceManager.getMyCertPath());
		
		PeerBalance oldBalance = null;
		if (remoteDNData != null) {
			oldBalance = AccountingControl.getInstance().getRemotePeerBalance(
					localDNData,
					remoteDNData,
					serviceManager.getDAO(AccountingDAO.class),
					serviceManager.getDAO(PeerDAO.class));
		}	
		
		if (oldBalance == null) {
			oldBalance = new PeerBalance(0., 0.);
		}
		
		double cpu = Math.max(oldBalance.getCPUTime() - accounting.getAccountings().getCPUTime(), 0);
		double data = Math.max(oldBalance.getData() - accounting.getAccountings().getData(), 0);
		AccountingControl.getInstance().setRemotePeerBalance(
				localDNData,
				remoteDNData, 
				new PeerBalance(cpu, data),
				serviceManager.getDAO(AccountingDAO.class),
				serviceManager.getDAO(PeerDAO.class));
	}

	/**
	 * Validates if the sender can save the balance and call the DAO to save it.
	 * 
	 * The sender (timer or component) must have the save key pair of the
	 * current module.
	 * 
	 * @param senderPublicKey The sender public key
	 */
	public void saveRanking() {
		String senderPublicKey = serviceManager.getSenderPublicKey();
		
		if (!serviceManager.isThisMyPublicKey(senderPublicKey)) {
			serviceManager.getLog().warn(AccountingMessages.getUnknownSenderSaveRankingMessage(senderPublicKey));
			return;
		}
		
		serviceManager.getLog().debug(AccountingMessages.getSaveRankingMessage());
		
		serviceManager.getDAO(AccountingDAO.class).saveBalancesRanking(
				serviceManager.getContainerContext().getProperty(PeerConfiguration.PROP_RANKINGFILE));
	}
}